package org.example.hrpc.registry.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.example.hrpc.common.URL;
import org.example.hrpc.common.constant.CommonConstant;
import org.example.hrpc.registry.ServiceMeta;
import org.example.hrpc.registry.AbstractRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;

public class ZookeeperRegistry  extends AbstractRegistry {
    private static  final Logger logger = LoggerFactory.getLogger(ZookeeperRegistry.class);
    private static final String root = CommonConstant.DEFAULT_NAME;

    String PATH_SEPARATOR = "/";
    private final CuratorFramework zkClient;

    private final ServiceDiscovery<ServiceMeta> serviceDiscovery;

    // url: registry://host:port/path?version=1.0?register=true
    public ZookeeperRegistry(URL url) {
        this.zkClient = CuratorFrameworkFactory.builder()
                .connectString(url.getSocketAddress().toString())
                // 会话超时
                .sessionTimeoutMs(5000)
                // 重试， 超时后1000ms重试一次
                .retryPolicy(new RetryOneTime(1000))
                .namespace(this.root)
                .build();

        this.zkClient.start();

        // 使用curator zk 本身提供的服务发现功能
        JsonInstanceSerializer<ServiceMeta> serializer = new JsonInstanceSerializer<>(ServiceMeta.class);
        this.serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMeta.class)
                .client(zkClient)
                .serializer(serializer)
                .basePath(this.root)
                .build();

        try {
            this.serviceDiscovery.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        logger.info("success connect to zookeeper: {}, ", url.getSocketAddress());
    }



//    @Override
//    public void doRegister(URL url) {
//        // 通过 Zookeeper 客户端创建节点，节点路径由 toUrlPath 方法生成，路径格式如下:
//        //   /${group}/${serviceInterface}/providers/${url}
//        // 比如
//        //   /dubbo/org.apache.dubbo.DemoService/providers/dubbo%3A%2F%2F127.0.0.1......
//        String path = toUrlPath(url);
//        try {
//            zkClient.create()
//                    .withMode(CreateMode.PERSISTENT)
//                    .forPath(path);
//        } catch (Exception e) {
//            throw new RuntimeException(e);
//        }
//    }
//
//    public void doUnregister(URL url) {
//        String path = toUrlPath(url);
//        try {
//            zkClient.delete()
//                    .forPath(path);
//        } catch (Exception e) {
//            throw new RuntimeException(e);
//        }
//    }



    private String toRootDir() {
        if (root.equals(PATH_SEPARATOR)) {
            return root;
        }
        return root + PATH_SEPARATOR;
    }

    private String toRootPath() {
        return root;
    }

    private String toServicePath(URL url) {
        String name = url.getPath();
        if ("*".equals(name)) {
            return toRootPath();
        }
        return toRootDir() + URL.encode(name);
    }


    private String toCategoryPath(URL url) {
        return toServicePath(url) + PATH_SEPARATOR + "providers";
    }

    private String toUrlPath(URL url) {
        return toCategoryPath(url) + PATH_SEPARATOR + URL.encode(url.toFullString());
    }

    @Override
    public void destroy() {
        super.destroy();
    }

    @Override
    public URL discovery(URL url) {
        URL targetURL = null;
        try {
            Collection<ServiceInstance<ServiceMeta>> serviceInstances = serviceDiscovery.queryForInstances(url.getPath());
            ServiceInstance<ServiceMeta> serviceInstance = LoadBalanceUtils.selector(new ArrayList<>(serviceInstances), "");
            // 根据返回值构造URL, 不要忘记拼接参数
            targetURL = new URL(CommonConstant.DEFAULT_NAME, serviceInstance.getAddress(), serviceInstance.getPort(),
                    serviceInstance.getName(), null);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return targetURL;
    }

    @Override
    protected void doRegister(URL url) {
        ServiceInstance<ServiceMeta> serviceInstance = null;
        try {
            serviceInstance = ServiceInstance.<ServiceMeta>builder()
                    .name(url.getPath())
                    .address(url.getHost())
                    .port(url.getPort())
                    .build();
            serviceDiscovery.registerService(serviceInstance);
            logger.info("url:{} success registers to zookeeper ", url);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    protected void doUnregister(URL url) {
        ServiceInstance<ServiceMeta> serviceInstance = null;
        try {
            serviceInstance = ServiceInstance.<ServiceMeta>builder()
                    .name(url.getPath())
                    .address(url.getHost())
                    .port(url.getPort())
                    .build();
            serviceDiscovery.unregisterService(serviceInstance);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
